package sqlengine.queryExecution;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.Map;
import java.util.Vector;
import java.util.Map.Entry;

import sqlengine.conditions.ConditionExecption;
import sqlengine.conditions.Conditions;
import sqlengine.conditions.JoinCondition;
import sqlengine.conditions.NonJoinCondition;
import sqlengine.conditions.Projection;
import sqlengine.database.DataManager;
import sqlengine.database.impl.Record;
import sqlengine.queryOptimization.PlanOptimizer;
import sqlengine.queryOptimization.QueryOptimizationException;

public class QueryExecution {
	PlanOptimizer planOptimizer;
	Conditions conditions;

	public QueryExecution(PlanOptimizer planOptimizer, Conditions conditions) {
		this.planOptimizer = planOptimizer;
		this.conditions = conditions;
	}

	public String execute() throws QueryOptimizationException,
			ConditionExecption, BasicOperationException {
		planOptimizer.checkConsistency();
		String[] indexCondition = planOptimizer.generateOptimizedPlan();
		ArrayList<Record> lhsTempResult = null;
		ArrayList<Record> rhsTempResult = null;
		ArrayList<Record> tempResult = null;
		ArrayList<JoinCondition> joinConditions = new ArrayList<JoinCondition>();
		ArrayList<NonJoinCondition> nonJoinConditions = new ArrayList<NonJoinCondition>();
		String retRestul = "";
		
		if (conditions.getTableNum() == 1) {
			if (indexCondition != null) {
				tempResult = BasicOperation.indexScan(new NonJoinCondition(
						conditions.getTableNames()[0], indexCondition[0],
						indexCondition[1]));
			} else {
				tempResult = BasicOperation.tableScan(conditions
						.getTableNames()[0]);
			}
			ArrayList<String> joinCond = conditions.getConditions().get(1);
			for (int i = 0; i < joinCond.size() / 2; i++) {
				joinConditions
						.add(new JoinCondition(conditions.getTableNames()[0],
								joinCond.get(2 * i),
								conditions.getTableNames()[0], joinCond
										.get(2 * i + 1)));
			}
			ArrayList<String> nonJoinCond = conditions.getConditions().get(0);
			for (int i = 0; i < nonJoinCond.size() / 2; i++) {
				nonJoinConditions.add(new NonJoinCondition(conditions
						.getTableNames()[0], nonJoinCond.get(2 * i),
						nonJoinCond.get(2 * i + 1)));
			}

			ArrayList<Record> finalResult = BasicOperation.filter(tempResult,
					joinConditions, nonJoinConditions, conditions
							.getTableNames());
			
			Iterator<Record> it = finalResult.iterator();
			int i = 1;
			while (it.hasNext()) {
				String tempStr = i++ + ":\n";
				Map<String, String> map = it.next().getRecord();
				
				if (conditions.isProjectAll()) {
					// project all
					Iterator<String> itrecord = DataManager.getOpenedFiles().get(conditions.getTableNames()[0]).getOrder().iterator();
					while (itrecord.hasNext()) {
						String column = itrecord.next();
						tempStr = tempStr
								+ (column + ": " + map.get(column))
								+ "\n";

					}
				} else {
					ArrayList<Projection> projections = conditions
							.getProjectionColumns();
					for (int j = 0; j < projections.size(); j++) {
						tempStr += projections.get(j).getTitle() + ": "
								+ map.get(projections.get(j).getColumnName())
								+ "\n";
					}
				}

				// tempStr += "\n";
				retRestul += tempStr + "\n";
			}
			if (i == 1) {
				retRestul = "no rows selected\n\n";
			}
		} else {

			if (indexCondition != null) {
				// 1. join with two index
				if (indexCondition.length == 4) {
					tempResult = BasicOperation
							.index_join_both(new JoinCondition(
									indexCondition[0], indexCondition[1],
									indexCondition[2], indexCondition[3]));
				} else if (indexCondition.length == 6) {
					// 2. nonjoin with two index
					lhsTempResult = BasicOperation
							.indexScan(new NonJoinCondition(indexCondition[0],
									indexCondition[1], indexCondition[2]));
					rhsTempResult = BasicOperation
							.indexScan(new NonJoinCondition(indexCondition[3],
									indexCondition[4], indexCondition[5]));
					// if (conditions.getConditions().get(2).size() != 0) {
					// tempResult = BasicOperation.nested_loops_join(
					// lhsTempResult, rhsTempResult, new JoinCondition(
					// conditions.getTableNames()[0],
					// conditions.getConditions().get(2).get(0),
					// conditions.getTableNames()[1],
					// conditions.getConditions().get(2).get(1)));
					// conditions.getConditions().remove(0);
					// conditions.getConditions().remove(0);
					// } else {
					tempResult = BasicOperation.cartesian_product(
							lhsTempResult, rhsTempResult);
					// }
				} else if (indexCondition.length == 5) {
					// 3. join with one index
					if (indexCondition[4].compareTo("first") == 0) {
						// first
						tempResult = BasicOperation.index_join(BasicOperation
								.tableScan(indexCondition[2]),
								new JoinCondition(indexCondition[0],
										indexCondition[1], indexCondition[2],
										indexCondition[3]), conditions
										.getTableNames());
					} else {
						// second
						tempResult = BasicOperation.index_join(BasicOperation
								.tableScan(indexCondition[0]),
								new JoinCondition(indexCondition[2],
										indexCondition[3], indexCondition[0],
										indexCondition[1]), conditions
										.getTableNames());
					}
				} else if (indexCondition.length == 3) {
					// 4. nonjoin with one index
					lhsTempResult = BasicOperation
							.indexScan(new NonJoinCondition(indexCondition[0],
									indexCondition[1], indexCondition[2]));
					if (indexCondition[0]
							.compareTo(conditions.getTableNames()[0]) == 0) {
						rhsTempResult = BasicOperation.tableScan(conditions
								.getTableNames()[1]);
					} else {
						rhsTempResult = BasicOperation.tableScan(conditions
								.getTableNames()[0]);
					}
					tempResult = BasicOperation.cartesian_product(
							lhsTempResult, rhsTempResult);
				}
			} else {
				// no index
				// two table scan
				tempResult = BasicOperation
						.cartesian_product(BasicOperation.tableScan(conditions
								.getTableNames()[0]), BasicOperation
								.tableScan(conditions.getTableNames()[1]));
			}

			// for non join condition
			ArrayList<String> nonjoinCond = conditions.getConditions().get(0);
			for (int j = 0; j < nonjoinCond.size() / 2; j++) {
				nonJoinConditions.add(new NonJoinCondition(conditions
						.getTableNames()[0], nonjoinCond.get(2 * j),
						nonjoinCond.get(2 * j + 1)));
			}
			nonjoinCond = conditions.getConditions().get(1);
			for (int j = 0; j < nonjoinCond.size() / 2; j++) {
				nonJoinConditions.add(new NonJoinCondition(conditions
						.getTableNames()[1], nonjoinCond.get(2 * j),
						nonjoinCond.get(2 * j + 1)));
			}

			// for join condition
			ArrayList<String> joinCond = conditions.getConditions().get(2);
			for (int j = 0; j < joinCond.size() / 2; j++) {
				joinConditions
						.add(new JoinCondition(conditions.getTableNames()[0],
								joinCond.get(2 * j),
								conditions.getTableNames()[1], joinCond
										.get(2 * j + 1)));
			}
			joinCond = conditions.getConditions().get(3);
			for (int j = 0; j < joinCond.size() / 2; j++) {
				joinConditions
						.add(new JoinCondition(conditions.getTableNames()[0],
								joinCond.get(2 * j),
								conditions.getTableNames()[0], joinCond
										.get(2 * j + 1)));
			}
			joinCond = conditions.getConditions().get(4);
			for (int j = 0; j < joinCond.size() / 2; j++) {
				joinConditions
						.add(new JoinCondition(conditions.getTableNames()[1],
								joinCond.get(2 * j),
								conditions.getTableNames()[1], joinCond
										.get(2 * j + 1)));
			}

			ArrayList<Record> finalResult = BasicOperation.filter(tempResult,
					joinConditions, nonJoinConditions, conditions
							.getTableNames());

			Iterator<Record> it = finalResult.iterator();
			int i = 1;

			ArrayList<Projection> projections;

			if (conditions.isProjectAll()) {
				projections = null;
			} else {
				projections = conditions.getProjectionColumns();
			}

			while (it.hasNext()) {
				String tempStr = "";
				tempStr = i++ + ":\n";
				Iterator<String> itrecord1, itrecord2;
				Map<String, String> map1, map2;

				itrecord1 = DataManager.getOpenedFiles().get(conditions.getTableNames()[0]).getOrder().iterator();
				itrecord2 = DataManager.getOpenedFiles().get(conditions.getTableNames()[1]).getOrder().iterator();
				map1 = it.next().getRecord();
				map2 = it.next().getRecord();
				
					
				if (projections == null) {
					while (itrecord1.hasNext()) {
						String column = itrecord1.next();
						tempStr += (column + ": " + map1.get(column))
								+ "\n";
					}
					while (itrecord2.hasNext()) {
							String column = itrecord2.next();
							tempStr += (column + ": " + map2.get(column))
									+ "\n";
					}
				} else {
					for (int k = 0; k < projections.size(); k++) {
						if (projections.get(k).getTableName().compareTo(
								conditions.getTableNames()[0]) == 0) {
							tempStr += projections.get(k).getTitle()
									+ ": "
									+ map1.get(projections.get(k)
											.getColumnName()) + "\n";
						} else {
							tempStr += projections.get(k).getTitle()
									+ ": "
									+ map2.get(projections.get(k)
											.getColumnName()) + "\n";
						}
					}
				}

				// map= it.next().getRecord();
				// itrecord = map.entrySet().iterator();

				// if (columns2 == null) {
				// while (itrecord.hasNext()) {
				// Entry<String, String> pairs;
				// pairs = (Entry<String, String>) itrecord.next();
				// tempStr += (pairs.getKey() + ": " + pairs.getValue())
				// + "\n";
				// }
				// } else {
				// for (int k = 0; k < columns2.size(); k++) {
				// tempStr += columns2.get(k) + ": "+
				// map.get(columns2.get(k))+"\n";
				// }
				// }

				// tempStr += "\n";
				retRestul += tempStr + "\n";
			}
			if (i == 1) {
				retRestul = "no rows selected\n\n";
			}

		}
		return retRestul.substring(0, retRestul.length()-1);
	}
}
